package com.asiainfo.dacp.datastash.broker.util;

import java.util.concurrent.CountDownLatch;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.leader.LeaderSelector;
import org.apache.curator.framework.recipes.leader.LeaderSelectorListenerAdapter;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.retry.RetryNTimes;
import org.apache.curator.utils.CloseableUtils;

import com.asiainfo.dacp.datastash.broker.core.BrokerContext;

/**
 *
 * 选主通用服务类
 *
 * @author MeiKefu
 * @date 2017-10-31
 */
public class LeaderElection {

    private static final String ZK_ADDRESS = BrokerContext.ZK_ADDRESS;
    private static final String ZK_PATH = "/datastash/broker/election";

    private LeaderSelector leaderSelector;
    private CuratorFramework client;
    private final CountDownLatch latch = new CountDownLatch(1);

    public LeaderElection(final LeaderSelectorListenerAdapter leaderSelectorListener) {
        client = CuratorFrameworkFactory.newClient(
                ZK_ADDRESS,
                new RetryNTimes(100, 5000)
        );
        leaderSelector = new LeaderSelector(client, ZK_PATH, new LeaderSelectorListenerAdapter(){
            @Override
            public void takeLeadership(CuratorFramework curatorFramework) throws Exception {
                leaderSelectorListener.takeLeadership(curatorFramework);
                
            }
            
            @Override
            public void stateChanged(CuratorFramework client, ConnectionState newState){
            	leaderSelectorListener.stateChanged(client, newState);
                if ( newState == ConnectionState.LOST ){
                	latch.countDown();
                }
            }
        });
//        leaderSelector.autoRequeue();
    }

    public void startup() throws InterruptedException {
        try{
            client.start();
            leaderSelector.start();
            latch.await(); 
        }finally{
        	CloseableUtils.closeQuietly(leaderSelector);
        	CloseableUtils.closeQuietly(client);
        }
    }
}
